import { usePlugin, HandlerMqConnection } from "./plugin";
import { Channel, Options, ConsumeMessage, GetMessage } from 'amqplib'

export function useMq<T>(handler: HandlerMqConnection<T>) {
    return usePlugin('@sker/mq', handler)
}
export interface IMqChannel<T> {
    (channel: Channel): Promise<T>;
}
export function useMqChannel<T>(sender: IMqChannel<T>) {
    return useMq(async (c) => {
        const channel = await c.createChannel()
        const res = await sender(channel)
        return res;
    })
}

export function useMqQueue<T>(queue: string, runner: IMqChannel<T>) {
    return useMqChannel(async c => {
        c.on('error', e => {
            console.error(e)
        })
        await c.assertQueue(queue, { durable: false })
        return await runner(c)
    })
}

export interface IMqConsume {
    (channel: Channel, message: ConsumeMessage | null): Promise<void>;
}
export function useMqConsume(queue: string, handler: IMqConsume, options?: Options.Consume) {
    return useMqQueue(queue, async (c) => {
        try {
            await c.prefetch(1, false)
            await c.consume(queue, (message) => {
                handler(c, message)
            }, { noAck: false, ...options })
        } catch (e) {
            console.error(e)
        }
    })
}

export function useMqConsumes(queue: string, handler: IMqConsume, prefetch: number = 1, options?: Options.Consume) {
    return useMqQueue(queue, async (c) => {
        await c.prefetch(prefetch, false)
        try {
            await c.consume(queue, async (message) => {
                await handler(c, message)
            }, { noAck: false, ...options })
        } catch (e) {
            console.error(e)
        }
    })
}


export interface IMqGet {
    (channel: Channel, message: GetMessage | null): Promise<void>;
}
export function useMqGet(queue: string, handler: IMqGet) {
    return useMqQueue(queue, async (c) => {
        const message = await c.get(queue, { noAck: false })
        if (message) {
            return await handler(c, message)
        }
    })
}
